package com.atguigu.gmall.realtime.util;

import com.atguigu.gmall.realtime.common.Constant;

/**
 * @Author lzc
 * @Date 2022/12/30 09:44
 */
public class SQLUtil {
    public static String getKafkaSourceDDL(String topic, String groupId, String ... dataFormat) {
    
        String format = "json";
        if (dataFormat.length > 0) {
            format = dataFormat[0];
        }
        
        return "with(" +
            " 'connector'='kafka', " +
            " 'properties.bootstrap.servers'='" + Constant.KAFKA_BROKERS + "', " +
            " 'properties.group.id'='" + groupId + "', " +
            " 'topic'='" + topic + "', " +
            " 'scan.startup.mode'='latest-offset', " +
            " 'format'='" + format + "' " +
            ")";
    }
    
    public static String getKafkaSinkDDL(String topic, String ... dataFormat) {
        String format = "json";
        if (dataFormat.length > 0) {
            format = dataFormat[0];
        }
        return "with(" +
            " 'connector'='kafka', " +
            " 'properties.bootstrap.servers'='" + Constant.KAFKA_BROKERS + "', " +
            " 'topic'='" + topic + "', " +
            " 'format'='" + format + "' " +
            ")";
    }
    
    public static String getUpsertKafkaDDL(String topic, String ... dataFormat) {
        String format = "json";
        if (dataFormat.length > 0) {
            format = dataFormat[0];
        }
        return "with(" +
            " 'connector'='upsert-kafka', " +
            " 'properties.bootstrap.servers'='" + Constant.KAFKA_BROKERS + "', " +
            " 'topic'='" + topic + "', " +
            " 'key.format'='" + format + "', " +
            " 'value.format'='" + format + "' " +
            ")";
    }
}
/*
key    value
a         10
a         null   表示: 删除前面的 a 10
a         20     新增
a         30     更新
a         null   删除     a         30


 */